package cn._51doit

import java.time.Duration
import java.util
import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerRecord, ConsumerRecords, KafkaConsumer}
import org.apache.kafka.common.serialization.StringDeserializer

/**
 * ┏┓　　　┏┓
 * ┏┛┻━━━┛┻┓
 * ┃　　　　　　　┃ 　
 * ┃　　　━　　　┃
 * ┃　┳┛　┗┳　┃
 * ┃　　　　　　　┃
 * ┃　　　┻　　　┃
 * ┃　　　　　　　┃
 * ┗━┓　　　┏━┛
 *    　　┃　　　┃神兽保佑
 *    　　┃　　　┃代码无BUG！
 *    　　┃　　　┗━━━┓
 *    　　┃　　　　　　　┣┓
 *    　　┃　　　　　　　┏┛
 *    　　┗┓┓┏━┳┓┏┛
 *    　　　┃┫┫　┃┫┫
 *    　　　┗┻┛　┗┻┛
 *
 * @create: 2021-10-12 20:14
 * @author: 今晚打脑斧
 * @program: ScalaCOnsumer
 * @Description:
 *      设置消费者不自动提交偏移量
 *      消费者默认值自动提交偏移量:enable.auto.commit 默认就是true
 *      自动提交偏移量的时间周期:auto.commit.interval.ms:默认是5000毫秒  interval:周期
 *      设置kafka不自动提交偏移量,以后我们编程可以精确控制
 *        enable.auto.commit: false
 **/
object ScalaConsumer2 {
  def main(args: Array[String]): Unit = {

    //指定broker的地址
    val properties = new Properties
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "doit01:9092,doit02:9092,doit03:9092")
    //指定key反序列化方式
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    //指定value反序列化方式
    properties.setProperty("value.deserializer", classOf[StringDeserializer].getName) // 两种写法都行
    //指定消费者的组ID
    properties.setProperty("group.id", "g0008")
    /*
    * 指定读取数据的偏移量
    * earliest：如果没有历史偏移量，从最开始读取数据，如果对应的这个组ID记录过偏移量，会接着偏移量继续读
    * latest: 消费者启动后产生的数据才读取（默认的）
    * none: 如果指定group.id没有历史偏移量，抛出异常
    */
    properties.setProperty("auto.offset.reset", "earliest")
    //消费者不自动提交偏移量
    properties.setProperty("enable.auto.commit", "false")


    //默认情况下，消费者消息完数据后，会自动更新偏移量
    val kafkaConsumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](properties)

    //读取数据
    val topics = util.Arrays.asList("wordcount")
    //订阅指定的topic，可以指定一个或多个topic
    kafkaConsumer.subscribe(topics)

    while (true) {

      val consumerRecords: ConsumerRecords[String, String] = kafkaConsumer.poll(Duration.ofSeconds(5))
      //println("----------------> " + System.currentTimeMillis())
      val iterator = consumerRecords.iterator()

      while (iterator.hasNext) {
        val record: ConsumerRecord[String, String] = iterator.next()

        //val line = record.value()
        println(record)
      }
    }

  }

}
